package org.example.state;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.state.*;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.runtime.state.FunctionInitializationContext;
import org.apache.flink.runtime.state.FunctionSnapshotContext;
import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.BroadcastConnectedStream;
import org.apache.flink.streaming.api.datastream.BroadcastStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction;
import org.apache.flink.util.Collector;
import org.example.data.WaterSensor;
import org.example.function.WaterSensorMapFunction;

/**
 * 水位超过阈值就告警
 */
public class BroadcastStateDemo {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(2);

        //数据流,只能读取广播状态
        SingleOutputStreamOperator<WaterSensor> sensorDS = env.socketTextStream("localhost", 9999)
                .map(new WaterSensorMapFunction());

        //1配置流（用来广播），广播流才能修改状态。
        DataStreamSource<String> thresholdDS = env.socketTextStream("localhost", 8888);

        //2将配置流广播出去
        MapStateDescriptor<String, Integer> broadcastState = new MapStateDescriptor<>("broadcast-state", Types.STRING, Types.INT);
        BroadcastStream<String> configBS = thresholdDS.broadcast(broadcastState);

        //3将数据库和广播后的配置流进行connect
        BroadcastConnectedStream<WaterSensor, String> broadcastCS = sensorDS.connect(configBS);

        SingleOutputStreamOperator<String> process = broadcastCS.process(new BroadcastProcessFunction<WaterSensor, String, String>() {
            @Override
            public void processElement(WaterSensor value, BroadcastProcessFunction<WaterSensor, String, String>.ReadOnlyContext ctx, Collector<String> out) throws Exception {
                //数据流的处理方法
                ReadOnlyBroadcastState<String, Integer> broadcastState1 = ctx.getBroadcastState(broadcastState);
                Integer threshold = broadcastState1.get("threshold");
                threshold = threshold == null ? 0 : threshold;
                if (value.getVc() > threshold) {
                    out.collect(value + "水位超过指定的阈值：" + threshold);
                }
            }

            @Override
            public void processBroadcastElement(String value, BroadcastProcessFunction<WaterSensor, String, String>.Context ctx, Collector<String> out) throws Exception {
                //广播后的配置流的处理方法
                //4获取广播状态,写入状态数据
                BroadcastState<String, Integer> broadcastState1 = ctx.getBroadcastState(broadcastState);
                broadcastState1.put("threshold", Integer.parseInt(value));
            }
        });

        process.print();


        env.execute("OperatorListState");
    }
}
